package net.zbus;

import java.util.HashSet;
import java.util.Set;

import net.zbus.ZMQ.Context;
import net.zbus.ZMQ.Socket;



/**
 * 异步worker,调用方负责维护心跳
 * 
 * @author jesse
 *
 *createtime 2013年11月14日 下午8:44:13 
 */
public class AsyncWorker implements IWorker {
	public static final String MODE_LB = "1"; 
	public static final String MODE_PUBSUB = "2";
	public static final String MODE_BC = "3";
	 
	private final String service;
	private final String mode;
	private final String registerToken;
	private final String accessToken;
	
	private final String host;
	private final int port;
	private final Context ctx;
	private Socket socket; 
	private Set<String> topics;
	private boolean registered = false;
	private long heartbeatAt = System.currentTimeMillis();
	
	
	private byte[] recvMsgId;  
	private byte[] recvSockId;
	
	
	private  ZMQ.Poller items;

	public AsyncWorker(WorkerConfig config){ 
		if(config.getCtx() == null){
			throw new IllegalArgumentException("context missing");
		}
		
		this.service = config.getService();
		this.mode = config.getMode();
		this.registerToken = config.getRegisterToken();
		this.accessToken = config.getAccessToken(); 
		this.ctx = config.getCtx();
		this.host = config.getHost();
		this.port = config.getPort();
		
		this.socket = ctx.socket(ZMQ.DEALER); 
		this.socket.setLinger(0);
		this.socket.setReceiveTimeOut(Protocol.HEARTBEAT_INTERVAL);
		
		this.items = new ZMQ.Poller(1);
        this.items.register(socket, ZMQ.Poller.POLLIN);
      
		
		String address = String.format("tcp://%s:%d", host, port);
		this.socket.connect(address); 
	}
	 
	private boolean sendCommand(String cmd, ZMsg args){
		if(args == null){
			args = new ZMsg();
		}
		args.pushFront(cmd); 
		args.pushFront(Protocol.MDPW);
		args.pushFront("");
		
		return args.send(this.socket);
	}
	private void register(){
		ZMsg msg = new ZMsg();
		msg.pushBack(this.service);
		msg.pushBack(this.registerToken);
		msg.pushBack(this.accessToken);
		msg.pushBack(this.mode);
		
		this.sendCommand(Protocol.MDPW_REG, msg);
		if(this.topics != null){
			//subscribe
			ZMsg topicArgs = new ZMsg();
			for(String topic : this.topics){ 
				topicArgs.pushBack(topic);
			}
			this.sendCommand(Protocol.MDPW_SUB, topicArgs);
		} 
		this.registered = true;
	}
	
	@Override
	public ZMsg recv(){ 
		if(!this.registered){
			this.register(); 
		}
		if(System.currentTimeMillis() > this.heartbeatAt){
			this.sendHeartbeat();
		} 
		
		if(this.items.poll()==-1){
			throw new ZBusException("unknow network error happened "); 
		}
		ZMsg res=null;
		if(this.items.pollin(0)){
			res = ZMsg.recvMsg(this.socket);
			if( res.size() < 3){
				throw new ZBusException("[<empty> <header>, <cmd>] frames required");
			}
			res.popFront(); //empty
			res.popFront(); //header
			String cmd = res.popFrontStr();
			
			if(Protocol.MDPW_JOB.equals(cmd)){
				if(res.size() < 2){
					throw new ZBusException("[<sock_id> <msg_id>] frames required");
				}
				this.recvSockId = res.unwrap().getData();
				this.recvMsgId = res.popFront();
				this.sendIdle();
				return res;
			} else if(Protocol.MDPW_HBT.equals(cmd)){
				//just do nothing for heartbeat
			} else if(Protocol.MDPW_DISC.equals(cmd)){ 
				throw new ZBusException(res.popFrontStr()); 
			} else if(Protocol.MDPW_SYNC.equals(cmd)){
				this.register();
			} else {
				throw new ZBusException("unknown worker command");
			} 
		}
		return res;

	}
	
	public void sendIdle(){
		this.sendCommand(Protocol.MDPW_IDLE, null);
	}
	
	public void sendHeartbeat(){
		this.sendCommand(Protocol.MDPW_HBT, null);
		this.heartbeatAt += Protocol.HEARTBEAT_INTERVAL;
	}
	
	public boolean send(ZMsg msg){ 
		if(this.recvSockId == null){
			throw new ZBusException("recvSockId should be set");
		}
		if(this.recvMsgId == null){
			throw new ZBusException("recvMsgId should be set");
		} 
		
		msg.pushFront(this.recvMsgId);
		msg.pushFront(Protocol.MDPC); 
		msg.pushFront("");
		msg.pushFront(this.recvSockId); 
		//msg.pushFront(Protocol.MDPW_REP);
		//msg.pushFront(Protocol.MDPW);
		msg.pushFront(Protocol.MDPX);
		msg.pushFront("");
		
		return msg.send(this.socket); 
	}
	
	public boolean subscribe(String... topic){
		if(!this.registered){
			this.register();
			this.registered = true;
		}
		if(this.topics == null){
			this.topics = new HashSet<String>();
		}
		ZMsg args = new ZMsg();
		for(String t : topic){
			this.topics.add(t);
			args.pushBack(t);
		}
		return this.sendCommand("", args);
	}
	
	public boolean unsubscribe(String... topic){
		if(!this.registered){
			this.register();
			this.registered = true;
		}
		if(this.topics == null){
			this.topics = new HashSet<String>();
		}
		ZMsg args = new ZMsg();
		for(String t : topic){
			this.topics.remove(t);
			args.pushBack(t);
		}
		return this.sendCommand(Protocol.MDPW_UNSUB, args);
	}
	
	public void destroy() {
		if (this.socket != null) {
			this.socket.close();
			this.socket = null;
		}
	}
	
	public static void main(String[] args) throws Exception {
		Context ctx = ZMQ.context(1);

		WorkerConfig config = new WorkerConfig();
		config.setCtx(ctx);
		
		config.setService("CONFIG_CENTER_PUB");
		config.setMode(MODE_PUBSUB);
		config.setHost("10.60.60.13");
		config.setPort(15555);
		AsyncWorker worker = new AsyncWorker(config);
		worker.subscribe("qheetest"); 
		while(true){
			ZMsg req = worker.recv();
			if(req == null){
				break;
			}
			req.dump();
			//ZMsg res = new ZMsg();
			//res.pushBack("this is frame pure java zbus");
			//worker.send(res);
		}
		worker.destroy();
		ctx.term();
		System.out.println("finish");
	}
	
}


